package ar.edu.itba.pod.legajo48062;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingQueue;

import ar.edu.itba.event.EventInformation;
import ar.edu.itba.event.RemoteEventDispatcher;
import ar.edu.itba.node.Node;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.node.api.ClusterAdministration;
import ar.edu.itba.pod.agent.runner.Agent;
import ar.edu.itba.pod.multithread.EventDispatcher;
import ar.edu.itba.pod.thread.CleanableThread;

public class DistributedEventDispatcher extends UnicastRemoteObject implements RemoteEventDispatcher {
	private static final long serialVersionUID = 1L;
	private WrapperEventDispatcher localDispatcher;
	private Cluster cluster;
	private ConcurrentMap<EventInformation, Long> processedEvents;
	private BlockingQueue<EventInformation> pendingEvents;
	private ConcurrentMap<NodeInformation, Long> nodesUpdate;
	
	public DistributedEventDispatcher(ClusterAdministration cluster, EventDispatcher localDispatcher) throws RemoteException{
		super();
		this.cluster=(Cluster) cluster;
		this.localDispatcher = (WrapperEventDispatcher) localDispatcher;
		
		pendingEvents = new LinkedBlockingQueue<EventInformation>();
		processedEvents = new ConcurrentHashMap<EventInformation, Long>();
		nodesUpdate = new ConcurrentHashMap<NodeInformation, Long>();
		
		new CleanableThread("processPendingEventThread") {
			public void run() {
				Cluster cluster = getCluster();
				while (!shouldFinish()){
					try {
						final EventInformation event;
						final Long currentTime;

						event = pendingEvents.take();
						currentTime = Host.getTime();
						processedEvents.put(event, currentTime);
						getLocalDispatcher().publish(event);
						
						new Communication<RemoteEventDispatcher>(cluster,Node.DISTRIBUTED_EVENT_DISPATCHER){
							@Override
							public Boolean remoteCallWithReturn(RemoteEventDispatcher stub, NodeInformation node) throws Exception {
								nodesUpdate.put(node, currentTime);
								return !stub.publish(event);
							}
						}.gossipBroadcast(null, true, true);
					} catch (InterruptedException e) {
						finish();
					}	
				}
			}
		}.start();
		
		new CleanableThread("cleanProcessedEventsThread") {
			public void run() {
				while (!shouldFinish()){
					Long currentTime = System.nanoTime();
					for (EventInformation event : processedEvents.keySet()){
						if (currentTime - processedEvents.get(event) > Host.eventOld ){
							processedEvents.remove(event);
						}
					}
					Host.sleep(Host.processedEventsRemoveFrequency);
				}
			};
		}.start();
		
		new CleanableThread("checkNewEventsThread") {
			public void run() {
				final Cluster cluster = getCluster();
				while (!shouldFinish()){
					new Communication<RemoteEventDispatcher>(cluster,Node.DISTRIBUTED_EVENT_DISPATCHER){
						@Override
						public void remoteCall(RemoteEventDispatcher stub, NodeInformation node) throws Exception {
							Set<EventInformation> newEvents = stub.newEventsFor(cluster.me());
							if (newEvents!=null){
								for (EventInformation event : newEvents){
									if (publish(event)){
										System.out.println("RECUPERE UN EVENTO QUE HABIA PERDIDO!!");
									}
								}
							}
						}
					}.broadcast(null,true, true);

					Host.sleep(Host.pullEventsFrequency);
				}
			};
		}.start();
	}
	
	@Override
	public BlockingQueue<Object> moveQueueFor(Agent agent)	throws RemoteException {
		return getLocalDispatcher().deregister(agent);
	}

	@Override
	public Set<EventInformation> newEventsFor(NodeInformation nodeInformation)	throws RemoteException {
		Set<EventInformation> newEvents = new HashSet<EventInformation>();
		Long lastUpdate = nodesUpdate.get(nodeInformation);
		
		for (EventInformation event : processedEvents.keySet() ){
			if (lastUpdate!=null){
				if (processedEvents.get(event)>lastUpdate){
					newEvents.add(event);
				}
			}
		}
		return newEvents;
	}

	@Override
	public boolean publish(EventInformation event) throws RemoteException, InterruptedException {
		synchronized (DistributedEventDispatcher.class) {
			if (pendingEvents.contains(event) || processedEvents.containsKey(event)){
				return false;
			}else{
				pendingEvents.put(event);
				return true;
			}
		}
	}
	
	public WrapperEventDispatcher getLocalDispatcher(){
		return localDispatcher;
	}

	public Cluster getCluster(){
		return cluster;
	}
}
